package org.example.infrastructure.persistent.repository;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.tuple.Pair;
import org.example.domain.user_raffle.model.entity.ActivityEntity;
import org.example.domain.user_raffle.model.entity.ActivityStrategyPrizeEntity;
import org.example.domain.user_raffle.model.entity.PrizeEntity;

import org.example.domain.user_raffle.model.entity.StockZeroTaskEntity;
import org.example.domain.user_raffle.repository.IUserRaffleRepo;
import org.example.domain.user_raffle.model.event.message.StockZeroMessage;
import org.example.infrastructure.persistent.dao.*;
import org.example.infrastructure.persistent.po.*;
import org.example.types.common.RedisKeysPrefix;
import org.redisson.api.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


/**
 * @author lxc18
 * @date 2025/3/30 9:37
 * @description UserRaffleRepo
 */
@Repository
public class UserRaffleRepo implements IUserRaffleRepo {
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private IUserActivityDao userActivityDao;
    @Resource
    private IActivityDao activityDao;
    @Resource
    private IActivityStrategyPrizeDao activityStrategyPrizeDao;
    @Resource
    private IPrizeDao iPrizeDao;
    @Resource
    private ILotteryRecordDao iLotteryRecordDao;
    @Resource
    private IStockZeroTaskDao iStockZeroTaskDao;

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Override
    public ActivityEntity queryActivityById(Long activityId) {
        //因为已经预热了，因此只从缓存里面找
        //先去缓存里面查找，如果缓存里面没有，再去数据库里面查找
        String cacheKey = RedisKeysPrefix.ACTIVITY_INFO + activityId;
        String jsonStr = (String) redissonClient.getBucket(cacheKey).get();
        ActivityEntity activityEntity = JSON.parseObject(jsonStr, ActivityEntity.class);
        if(activityEntity == null){
            ActivityPO po = activityDao.queryActivityById(activityId);
            if(po == null){
                return null;
            }
            activityEntity = ActivityEntity.builder()
                    .activityId(po.getActivityId())
                    .activityName(po.getActivityName())
                    .startTime(po.getStartTime())
                    .endTime(po.getEndTime())
                    .createTime(po.getCreateTime())
                    .updateTime(po.getUpdateTime())
                    .build();
            //将活动信息放入缓存
            redissonClient.getBucket(cacheKey).set(JSON.toJSONString(activityEntity));
        }
        return activityEntity;
    }

    @Override
    public int queryUserQuota(Long activityId, Long userId) {
        String key = RedisKeysPrefix.USER_PARTICIPATE_QUOTA+userId;

        RMap<Long, Integer> map = redissonClient.getMap(key);
        Integer quota = map.get(activityId);
        if(quota != null){
            quota = (int) quota;
        }
        else{
            //调用mapper方法
            quota = userActivityDao.queryUserParticipateQuota(userId,activityId);
            map.put(activityId, quota);
        }
        //        查看redis所有key

        return quota;
    }

    @Override
    public int queryUserUsedQuota(Long activityId, Long userId) {
        String key = RedisKeysPrefix.USER_PARTICIPATE_USED_QUOTA+userId;
        RMap<Long, Integer> map = redissonClient.getMap(key);
        Integer quota = map.get(activityId);
        if(quota != null){
            quota = (int) quota;
        }
        else{
            //调用mapper方法
            quota = userActivityDao.queryUserUsedQuota(userId,activityId);
            map.put(activityId, quota);
        }
        //        查看redis所有key
        return quota;
    }

    @Override
    public boolean isInBlackList(Long activityId, Long userId) {
        String key = RedisKeysPrefix.USER_PARTICIPATE_IS_BLACK+userId;
        RMap<Long, Integer> map = redissonClient.getMap(key);
        Integer isBlack = map.get(activityId);
        if (isBlack==null){
            isBlack = userActivityDao.queryUserIsBlack(userId, activityId);
            map.put(activityId,isBlack);
        }
        return isBlack==1;
    }

    @Override
    public ActivityStrategyPrizeEntity queryStrategy(Long activityId, String strategyName) {
        //先查缓存，再查数据库
        String cacheKey = RedisKeysPrefix.ACTIVITY_STRATEGY + activityId;
        RMap<String, String> map = redissonClient.getMap(cacheKey);
        String jsonStr = map.get(strategyName);
        ActivityStrategyPrizeEntity activityStrategyPrizeEntity = JSON.parseObject(jsonStr, ActivityStrategyPrizeEntity.class);;
        if(activityStrategyPrizeEntity != null){
            return activityStrategyPrizeEntity;
        }
        //查数据库
        ActivityStrategyPrizePO activityStrategyPrizePO = activityStrategyPrizeDao.queryStrategy(activityId, strategyName);
        activityStrategyPrizeEntity = new ActivityStrategyPrizeEntity();
        activityStrategyPrizeEntity.setActivityId(activityStrategyPrizePO.getActivityId());
        activityStrategyPrizeEntity.setStrategyName(activityStrategyPrizePO.getStrategyName());
        activityStrategyPrizeEntity.setPrizeIds(activityStrategyPrizePO.getPrizeIds());
        activityStrategyPrizeEntity.setConfig(activityStrategyPrizePO.getConfig());
        //存入缓存
        map.put(strategyName, JSON.toJSONString(activityStrategyPrizeEntity));
        return activityStrategyPrizeEntity;
    }

    @Override
    public PrizeEntity queryPrizeById(Long prizeId) {
        String cacheKey = RedisKeysPrefix.PRIZE_INFO + prizeId;
        String jsonStr= (String) redissonClient.getBucket(cacheKey).get();
        PrizeEntity prizeEntity = JSON.parseObject(jsonStr, PrizeEntity.class);
        if (prizeEntity == null) {
            PrizePO prizePO = iPrizeDao.queryPrizeById(prizeId);
            if (prizePO != null) {
                prizeEntity = PrizeEntity.builder()
                        .prizeId(prizePO.getPrizeId())
                        .prizeName(prizePO.getPrizeName())
                        .prizeDescription(prizePO.getPrizeDescription())
                        .prizeType(prizePO.getPrizeType())
                        .probability(prizePO.getProbability())
                        .stockAvailable(prizePO.getStockAvailable())
                        .stockUsed(prizePO.getStockUsed())
                        .createTime(prizePO.getCreateTime())
                        .updateTime(prizePO.getUpdateTime())
                        .build();
                // 缓存奖品信息
                redissonClient.getBucket(cacheKey).set(JSON.toJSONString(prizeEntity));
            }
        }
        return prizeEntity;
    }

    @Override
    public int queryPrizeStock(Long prizeId) {
        //缓存是有可能被清理的，因此需要从数据库里面查
        String cacheKey = RedisKeysPrefix.PRIZE_STOCK + prizeId;
        RAtomicLong stock = redissonClient.getAtomicLong(cacheKey);
        // 加分布式锁防止缓存击穿
        if (!stock.isExists()) {
            boolean lockAcquired = redissonClient.getLock("init_stock_lock:" + prizeId).tryLock();
            if (lockAcquired) {
                try {
                    PrizePO prizePO = iPrizeDao.queryPrizeById(prizeId);
                    if (prizePO != null) {
                        stock.set(prizePO.getStockAvailable());
                    }
                } finally {
                    redissonClient.getLock("init_stock_lock:" + prizeId).unlock();
                }
            }
        }
        return (int) stock.get();
    }

    @Override
    public Long deductPrizeStock(Long prizeId) {
        //分布式扣减库存，返回分三种，扣减成功，剩余库存＞0;扣减成功，剩余库存=0;扣减失败，库存不足
        String cacheKey = RedisKeysPrefix.PRIZE_STOCK + prizeId;
        String cacheKey_usedStock = RedisKeysPrefix.PRIZE_USED_STOCK + prizeId;
        // 执行 Lua 脚本，返回值为脚本执行的结果
        RScript script = redissonClient.getScript();

        // 脚本执行结果
        Long result = script.eval(RScript.Mode.READ_WRITE,
                RedisKeysPrefix.LUA_SCRIPT_STOCK,
                RScript.ReturnType.INTEGER,
                Arrays.asList(cacheKey, cacheKey_usedStock), // 传递两个 Redis 键
                Collections.emptyList()); // 如果有额外参数，可以在这里传

        System.out.println("result = " + result);
        // 返回值判断：
        // -1 表示库存不足
        // 1 表示扣减成功，剩余库存 > 0
        // 0 表示扣减成功，剩余库存 = 0
        //
        return result;



    }

    @Override
    public boolean addUserRaffleRecord(Long activityId, Long userId, Long prizeId) {
        //TODO: 发送消息到mq，添加抽奖记录,写到数据库中
        LotteryRecordPO lotteryRecordPO = new LotteryRecordPO();
        lotteryRecordPO.setActivityId(activityId);
        lotteryRecordPO.setUserId(userId);
        lotteryRecordPO.setPrizeId(prizeId);
//        lotteryRecordPO.setDistributeTime(LocalDateTime.now());
        lotteryRecordPO.setStatus("CREATED");

        return iLotteryRecordDao.insert(lotteryRecordPO);
    }

    @Override
    public List<Long> getAllPrizeIds() {
        //先从缓存里面查
        String cacheKey = RedisKeysPrefix.PRIZE_INFO;
        //redissonClient查询前缀为cacheKey的所有key
        RKeys keys = redissonClient.getKeys();
        // 修正：将 Iterable<String> 转换为 List<String>
        List<Long> res = new ArrayList<>();
        Iterable<String> keysIterable = keys.getKeysByPattern(cacheKey + "*");
        for (String key : keysIterable) {  // 正确遍历 Iterable<String>
            String prizeId = key.substring(cacheKey.length());
            res.add(Long.parseLong(prizeId));
        }
        return res;
    }

    @Override
    public void decrPrizeStock(Long prizeId,int size) {
        iPrizeDao.decrPrizeStock(prizeId,size);
    }

    @Override
    public int deductUserQuotaAndAddUsedQuota(Long activityId, Long userId) {
        //一个用户同时参加一个活动，可能会有多个线程同时扣减用户额度，因此需要加锁
        //加分布式锁防止缓存击穿
        // 使用更细粒度的锁，并确保锁的自动释放
        RLock lock = redissonClient.getLock("deduct_quota_lock:" + activityId + ":" + userId);
        boolean lockAcquired = false;
        try {
            // 设置最大等待时间为0（立即返回），锁持有时间30秒防止死锁
            lockAcquired = lock.tryLock(0, 30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return -1; // 可根据业务需求返回特定中断代码
        }

        if (!lockAcquired) {
            return -1; // 快速失败，避免阻塞
        }

        try {
            // 统一维护缓存键名
            final String cacheKeyQuota = RedisKeysPrefix.USER_PARTICIPATE_QUOTA + userId;
            final String cacheKeyUsed = RedisKeysPrefix.USER_PARTICIPATE_USED_QUOTA + userId;

            // 使用自动加载缓存的原子操作
            RMap<Long, Integer> quotaMap = redissonClient.getMap(cacheKeyQuota);
            RMap<Long, Integer> usedQuotaMap = redissonClient.getMap(cacheKeyUsed);

            // 使用computeIfAbsent实现原子化的缓存加载
            Integer quota = quotaMap.computeIfAbsent(activityId, k ->
                    userActivityDao.queryUserParticipateQuota(userId, activityId)
            );
            Integer usedQuota = usedQuotaMap.computeIfAbsent(activityId, k ->
                    userActivityDao.queryUserUsedQuota(userId, activityId)
            );

            // 防御空指针（数据库无记录情况）
            if (quota == null || usedQuota == null) {
                return -1; // 或定义业务错误码
            }

            // 额度不足时提前返回
            if (quota <= 0) {
                return -1;
            }

            // 使用Redis管道批量提交更新
            RBatch batch = redissonClient.createBatch();
            batch.getMap(cacheKeyQuota).putAsync(activityId, quota - 1);
            batch.getMap(cacheKeyUsed).putAsync(activityId, usedQuota + 1);
            batch.execute();

            // 返回操作后的最新已用额度
            return usedQuota + 1;
        } catch (Exception e) {
            // 日志记录异常信息
//            logger.error("Deduct quota failed for user:{} activity:{}", userId, activityId, e);
            return -1; // 或定义系统错误码
        } finally {
            if (lockAcquired) {
                lock.unlock(); // 确保锁最终释放
            }
        }
    }

    @Override
    public void clearUsedQuota(Long activityId, Long userId,int quota) {
        String cacheKey_usedQuota = RedisKeysPrefix.USER_PARTICIPATE_USED_QUOTA + userId;
        RMap<Long, Integer> usedQuotaMap = redissonClient.getMap(cacheKey_usedQuota);
        //最新的值减去这个值
        usedQuotaMap.addAndGet(activityId, -quota);
    }

    @Override
    public Long postProcess(Long userId, Long activityId, Long result, Long prizeId) {
        if (result == 0) {
            //TODO: 发送消息到mq异步更新数据库
            //创建消息对象，并添加到数据库
            StockZeroTaskPO stockZeroTaskPO = new StockZeroTaskPO();
            //uuid，唯一id
            stockZeroTaskPO.setPrizeId(prizeId);
            stockZeroTaskPO.setStatus("CREATED");
            iStockZeroTaskDao.insert(stockZeroTaskPO);
            // 获取插入后的自增 ID
            Long id = stockZeroTaskPO.getId();
            StockZeroMessage message = new StockZeroMessage(id, prizeId);
            String JSONMessage = JSON.toJSONString(message);
            try {
                //发送mq消息
                rabbitTemplate.convertAndSend("stock-zero-exchange","stock.zero",JSONMessage);
                //数据库消息状态更新为成功
            }
            catch (Exception e){
                //TODO: 异步更新数据库失败
                //数据库消息状态更新为失败
                iStockZeroTaskDao.updateStatusToFailed(id);

            }


        }
        else if ( result == 1) {
            String cacheKey_delay = RedisKeysPrefix.PRIZE_DELAY_QUEUE + prizeId;
            //将奖品id添加到延迟队列中
            RQueue<Object> queue = redissonClient.getBlockingQueue(cacheKey_delay);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(queue);
            delayedQueue.offer(prizeId, 3, TimeUnit.SECONDS);

        }
        else if ( result == -1) {
            prizeId = null;
        }
        return prizeId;
    }

    @Override
    public void updateStatusSuccess(Long id) {
        iStockZeroTaskDao.updateStatusToSuccess(id);
    }

    @Override
    public boolean isTaskStockZero(Long id) {
        StockZeroTaskPO stockZeroTaskPO = iStockZeroTaskDao.queryById(id);
        if (stockZeroTaskPO!= null && stockZeroTaskPO.getStatus().equals("SUCCESS")) {
            return true;
        }
        return false;
    }

    @Override
    public void updateStockToZero(Long prizeId) {

        //清空奖品对应的延迟队列
        String cacheKey_delay = RedisKeysPrefix.PRIZE_DELAY_QUEUE + prizeId;
        RQueue<Object> queue = redissonClient.getBlockingQueue(cacheKey_delay);
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(queue);
        delayedQueue.clear();
        // 解绑延迟队列（防止残留）
        delayedQueue.destroy();
        // 删除原始队列（Redis 中的 key）
        queue.delete();
        //将数据库中奖品库存更新为0
        iPrizeDao.updateStockToZero(prizeId);
    }

    @Override
    public List<StockZeroTaskEntity> getAllFailedTaskIds() {
        List<StockZeroTaskPO> stockZeroTaskPOList = iStockZeroTaskDao.queryAllFailedTaskIds();
        List<StockZeroTaskEntity> stockZeroTaskEntityList = new ArrayList<>();
        for (StockZeroTaskPO stockZeroTaskPO : stockZeroTaskPOList) {
            StockZeroTaskEntity stockZeroTaskEntity = StockZeroTaskEntity.builder()
                        .id(stockZeroTaskPO.getId())
                        .prizeId(stockZeroTaskPO.getPrizeId())
                        .status(stockZeroTaskPO.getStatus())
                        .createTime(stockZeroTaskPO.getCreateTime())
                        .updateTime(stockZeroTaskPO.getUpdateTime())
                        .build();
            stockZeroTaskEntityList.add(stockZeroTaskEntity);
        }
        return stockZeroTaskEntityList;
    }

    @Override
    public boolean isGuaranteePrize(Long activityId, Long prizeId) {
        String cacheKey = RedisKeysPrefix.ACTIVITY_STRATEGY + activityId;
        RMap<String, String> map = redissonClient.getMap(cacheKey);
        String jsonStr = map.get("GUARANTEE");
        ActivityStrategyPrizeEntity activityStrategyPrizeEntity = JSON.parseObject(jsonStr, ActivityStrategyPrizeEntity.class);
        if (activityStrategyPrizeEntity!= null && activityStrategyPrizeEntity.getPrizeIds().contains(prizeId.toString())) {
            return true;
        }
        return false;
    }

    @Override
    public List<Pair<Long, BigDecimal>> queryPrizeByIds(List<Long> prizeIds) {
        List<Pair<Long, BigDecimal>> resultList = new ArrayList<>();
        if (CollectionUtils.isEmpty(prizeIds)) {
            return resultList;
        }

        // 构造缓存 key 列表
        List<String> cacheKeys = prizeIds.stream()
                .map(id -> RedisKeysPrefix.PRIZE_INFO + id)
                .collect(Collectors.toList());

        // 批量获取缓存
        RBatch batch = redissonClient.createBatch();
        List<RFuture<Object>> futureList = new ArrayList<>();
        for (String key : cacheKeys) {
            futureList.add(batch.getBucket(key).getAsync());
        }
        batch.execute();

        // 处理缓存命中的数据
        Map<Long, PrizeEntity> cachedPrizeMap = new HashMap<>();
        for (int i = 0; i < futureList.size(); i++) {
            Object val = futureList.get(i).getNow();
            if (val != null) {
                PrizeEntity prize = JSON.parseObject(val.toString(), PrizeEntity.class);
                cachedPrizeMap.put(prizeIds.get(i), prize);
            }
        }

        // 找出未命中的 prizeIds
        List<Long> missedIds = prizeIds.stream()
                .filter(id -> !cachedPrizeMap.containsKey(id))
                .collect(Collectors.toList());

        // 回源数据库
        if (!missedIds.isEmpty()) {
            List<PrizePO> prizePOList = iPrizeDao.queryPrizeByIds(missedIds); // 你需要实现这个批量查询方法
            for (PrizePO po : prizePOList) {
                PrizeEntity entity = PrizeEntity.builder()
                        .prizeId(po.getPrizeId())
                        .prizeName(po.getPrizeName())
                        .prizeDescription(po.getPrizeDescription())
                        .prizeType(po.getPrizeType())
                        .probability(po.getProbability())
                        .stockAvailable(po.getStockAvailable())
                        .stockUsed(po.getStockUsed())
                        .createTime(po.getCreateTime())
                        .updateTime(po.getUpdateTime())
                        .build();

                cachedPrizeMap.put(po.getPrizeId(), entity);
                // 回写缓存（异步写也可以）
                redissonClient.getBucket(RedisKeysPrefix.PRIZE_INFO + po.getPrizeId()).set(JSON.toJSONString(entity));
            }
        }

        // 整理结果
        for (Long prizeId : prizeIds) {
            PrizeEntity entity = cachedPrizeMap.get(prizeId);
            if (entity != null) {
                resultList.add(Pair.of(prizeId, BigDecimal.valueOf(entity.getProbability())));
            }
        }

        return resultList;
    }

    @Override
    public Map<String, ActivityStrategyPrizeEntity> queryAllStrategy(Long activityId) {
        String cacheKey = RedisKeysPrefix.ACTIVITY_STRATEGY + activityId;
        RMap<String, String> map = redissonClient.getMap(cacheKey);

        // 如果缓存中已经有值，直接构造返回
        Map<String, ActivityStrategyPrizeEntity> strategyMap = new HashMap<>();
        if (!map.isEmpty()) {

            for (Map.Entry<String, String> entry : map.entrySet()) {
                strategyMap.put(entry.getKey(), JSON.parseObject(entry.getValue(), ActivityStrategyPrizeEntity.class));
            }
            return strategyMap;
        }

        // 缓存未命中，查询数据库
        // 写入缓存并构造返回结果
        List<ActivityStrategyPrizePO> strategyList = activityStrategyPrizeDao.queryStrategyByActivityId(activityId);
        for (ActivityStrategyPrizePO po : strategyList) {
            ActivityStrategyPrizeEntity entity = ActivityStrategyPrizeEntity.builder()
                        .activityId(po.getActivityId())
                        .strategyName(po.getStrategyName())
                        .prizeIds(po.getPrizeIds())
                        .config(po.getConfig())
                        .build();
            strategyMap.put(po.getStrategyName(), entity);
            map.put(po.getStrategyName(), JSON.toJSONString(entity));
        }
        return strategyMap;
    }



    @Override
    public void updateStatusFailed(Long id) {
        iStockZeroTaskDao.updateStatusToFailed(id);
    }
}
